#include <stdafx.h>

#include <cstdarg>

#include "PC2DataStore.h"


namespace PC2
{

class KeyFormatBuffer
{
public:
	int Scanf(char const* apFormat, ...)
	{
		va_list args;
		va_start (args, apFormat);
		int ret = vsprintf(m_Buffer, apFormat, args);
		va_end (args);
		return ret;
	}

public:
	char m_Buffer[64];
};

PC2DataStore::PC2DataStore(char const* apPath)
{
	leveldb::Options lOpenOptions;
	lOpenOptions.create_if_missing = true;
	leveldb::Status lStatus;

	lStatus = leveldb::DB::Open(lOpenOptions, apPath, &m_pDB);

	if(false == lStatus.ok())
	{
		LOG("open db %s error %s", apPath, lStatus.ToString().c_str());
	}
}

PC2DataStore::~PC2DataStore()
{

}

void PC2DataStore::WriteCurrentTransId(leveldb::WriteBatch* apBatch, uint32_t aTransId)
{
	apBatch->Put("_CURR_ID", leveldb::Slice((char*)&aTransId, sizeof(uint32_t)));
}

void PC2DataStore::WriteTransData(leveldb::WriteBatch* apBatch, TransactionInfoSet* apInfoSet)
{
	KeyFormatBuffer lBuffer;
	lBuffer.Scanf("#%u_DATA", apInfoSet->m_LastTrans.m_TransId);
	apBatch->Put(lBuffer.m_Buffer, leveldb::Slice(apInfoSet->m_TransData.c_str(), apInfoSet->m_TransData.length()));
}

void PC2DataStore::WriteLastTransState( leveldb::WriteBatch* apBatch, TransactionInfoSet* apInfoSet)
{
	KeyFormatBuffer lBuffer;
	lBuffer.Scanf("#%u_LASTTRANS", apInfoSet->m_LastTrans.m_TransId);
	apBatch->Put(lBuffer.m_Buffer, leveldb::Slice((char*)&apInfoSet->m_LastTrans, sizeof(Trans)));
}

void PC2DataStore::WritePrepareResponseSet(leveldb::WriteBatch* apBatch, TransactionInfoSet* apInfoSet)
{
	KeyFormatBuffer lBuffer;
	lBuffer.Scanf("#%u_PRE_RESP", apInfoSet->m_LastTrans.m_TransId);
	std::string lData;
	for (auto it = apInfoSet->m_PrepareResponses.begin();it != apInfoSet->m_PrepareResponses.end(); it++)
	{
		lData.append((char*)&(it->second),sizeof(Trans));
	}
	apBatch->Put(lBuffer.m_Buffer, lData);
}


void PC2DataStore::WriteRelatedNodeSet( leveldb::WriteBatch* apBatch, TransactionInfoSet* apInfoSet )
{
	KeyFormatBuffer lBuffer;
	lBuffer.Scanf("#%u_NODESET", apInfoSet->m_LastTrans.m_TransId);
	std::string lData;
	for (auto it = apInfoSet->m_RelatedNodes.begin();it != apInfoSet->m_RelatedNodes.end(); it++)
	{
		lData.append((char*)&(*it),sizeof(uint32_t));
	}
	apBatch->Put(lBuffer.m_Buffer, lData);
}

void PC2DataStore::FinishAndClearTransactionInfo( leveldb::WriteBatch* apBatch, TransactionInfoSet* apInfoSet )
{
	KeyFormatBuffer lBuffer;
	
	if(false == apInfoSet->m_TransData.empty())
	{
		lBuffer.Scanf("#%u_DATA", apInfoSet->m_LastTrans.m_TransId);
		apBatch->Delete(lBuffer.m_Buffer);
	}

	if(false == apInfoSet->m_RelatedNodes.empty())
	{
		lBuffer.Scanf("#%u_NODESET", apInfoSet->m_LastTrans.m_TransId);
		apBatch->Delete(lBuffer.m_Buffer);
	}

	if(false == apInfoSet->m_PrepareResponses.empty())
	{
		lBuffer.Scanf("#%u_PRE_RESP", apInfoSet->m_LastTrans.m_TransId);
		apBatch->Delete(lBuffer.m_Buffer);
	}
}

std::uint32_t PC2DataStore::GetCurrentTransId()
{
	std::string lValue;
	leveldb::Status lStatus = m_pDB->Get(leveldb::ReadOptions(), "_CURR_ID" ,&lValue);

	if(lStatus.ok())
	{
		return *(uint32_t*)lValue.c_str();
	}
	else
	{
		return 0;
	}
}


bool PC2DataStore::LoadTranInfoSet(uint32_t aTransID, TransactionInfoSet* apSet, bool aCollectResponse)
{
	std::string lValue;
	leveldb::Status lStatus;
	KeyFormatBuffer lBuffer;

	//last trans state
	lBuffer.Scanf("#%u_LASTTRANS", aTransID);
	lStatus = m_pDB->Get(leveldb::ReadOptions(), lBuffer.m_Buffer ,&lValue);
	if(lStatus.ok())
	{
		apSet->m_LastTrans = *(Trans*)lValue.c_str();
	}
	else
	{
		return false;
	}

	//data
	lBuffer.Scanf("#%u_DATA", aTransID);
	lStatus = m_pDB->Get(leveldb::ReadOptions(), lBuffer.m_Buffer , &apSet->m_TransData);


	//should return?
	if(false == aCollectResponse)
	{
		return true;
	}

	//related nodes
	lBuffer.Scanf("#%u_NODESET", aTransID);
	lStatus = m_pDB->Get(leveldb::ReadOptions(), lBuffer.m_Buffer ,&lValue);
	if(lStatus.ok())
	{
		apSet->m_RelatedNodes.clear();
		for (uint32_t i = 0;i<lValue.length(); i+= sizeof(uint32_t))
		{
			uint32_t lTrans = *(uint32_t*)(lValue.c_str() + i);
			apSet->m_RelatedNodes.insert(lTrans);
		}
	}

	// prepare resp
	lBuffer.Scanf("#%u_PRE_RESP", aTransID);
	lStatus = m_pDB->Get(leveldb::ReadOptions(), lBuffer.m_Buffer, &lValue);

	if(lStatus.ok())
	{
		apSet->m_PrepareResponses.clear();
		for (uint32_t i = 0;i<lValue.length(); i+= sizeof(Trans))
		{
			Trans lTrans = *(Trans*)(lValue.c_str() + i);
			apSet->m_PrepareResponses.insert(std::make_pair(lTrans.m_NodeId, lTrans));
		}
	}

	return true;
}

void PC2DataStore::CommitBatch(leveldb::WriteBatch* apBatch)
{
	leveldb::Status lStatus = m_pDB->Write(leveldb::WriteOptions(), apBatch);
	if(false == lStatus.ok())
	{
		LOG("write db batch %s error %s", lStatus.ToString().c_str());
	}
}

leveldb::Iterator* PC2DataStore::CreateIterator()
{
	return m_pDB->NewIterator(leveldb::ReadOptions());
}



}
